{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<p><font size=-1 color=gray>\n",
    "&copy; Copyright 2018 IBM Corp. All Rights Reserved.\n",
    "<p>\n",
    "Licensed under the Apache License, Version 2.0 (the \"License\"); you may not use this file\n",
    "except in compliance with the License. You may obtain a copy of the License at\n",
    "http://www.apache.org/licenses/LICENSE-2.0\n",
    "Unless required by applicable law or agreed to in writing, software distributed under the\n",
    "License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either\n",
    "express or implied. See the License for the specific language governing permissions and\n",
    "limitations under the License.\n",
    "</font></p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analyze Clickstream Events\n",
    "\n",
    "This notebook uses the [Scala](https://www.scala-lang.org/) programming language\n",
    "to interact with IBM Db2 Event Stream. It demonstrates how to:\n",
    "\n",
    "* Connect to Event Store\n",
    "* Analyze clickstream data to gain insight into customer interests\n",
    "* Visualize the information with interactive Brunel charts"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect to IBM Db2 Event Store\n",
    "\n",
    "### Determine the IP address of your host\n",
    "\n",
    "Obtain the IP address of the host that you want to connect to by running the appropriate command for your operating system:\n",
    "\n",
    "* On Mac, run: `ifconfig`\n",
    "* On Windows, run: `ipconfig`\n",
    "* On Linux, run: `hostname -i`\n",
    "\n",
    "Edit the `HOST = \"XXX.XXX.XXX.XXX\"` value in the next cell to provide the IP address."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// Set your host IP address\n",
    "val Host = \"XXX.XXX.XXX.XXX\"\n",
    "\n",
    "// Port will be 1100 for version 1.1.2 or later (5555 for version 1.1.1)\n",
    "val Port = \"1100\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Add Brunel integration\n",
    "Use cell magic to install the Brunel integration for Apache Toree (Scala)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%AddJar -magic https://brunelvis.org/jar/spark-kernel-brunel-all-2.3.jar -f"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import Scala packages\n",
    "\n",
    "Import packages for Scala, Spark, and IBM Db2 Event Store."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import sys.process._\n",
    "import java.io.File\n",
    "import scala.concurrent.{Await, Future}\n",
    "import scala.concurrent.duration.Duration\n",
    "import org.apache.log4j.{Level, LogManager, Logger}\n",
    "import org.apache.spark._\n",
    "import org.apache.spark.sql.expressions.Window\n",
    "import org.apache.spark.sql.functions._\n",
    "import org.apache.spark.sql.ibm.event.EventSession\n",
    "import org.apache.spark.sql.Row\n",
    "import org.apache.spark.sql.types._\n",
    "import com.ibm.event.catalog.TableSchema\n",
    "import com.ibm.event.common.ConfigurationReader\n",
    "import com.ibm.event.example.DataGenerator\n",
    "import com.ibm.event.oltp.EventContext\n",
    "import com.ibm.event.oltp.InsertResult"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect to Event Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ConfigurationReader.setConnectionEndpoints(Host + \":\" + Port)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load data from the Event Store table into a DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val sqlContext = new EventSession(spark.sparkContext, \"TESTDB\")\n",
    "import sqlContext.implicits._\n",
    "\n",
    "val table = sqlContext.loadEventTable(\"ClickStreamTable\")\n",
    "table.registerTempTable(\"ClickStreamTable\")\n",
    "\n",
    "val clickStreamDF = sqlContext.sql(\"select * from ClickStreamTable\")\n",
    "clickStreamDF.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare the clickstream data\n",
    "\n",
    "Use Spark SQL and Spark functions to build DataFrames with aggregated web metrics.\n",
    "\n",
    "### Calculate time spent on web pages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val timestamp = clickStreamDF(\"timestamp\")\n",
    "val next_timestamp = lead(timestamp, 1).over(Window.orderBy(timestamp))\n",
    "\n",
    "// Calculate time on spent on web pages\n",
    "val clickStreamWithTimeDF = clickStreamDF.withColumn(\n",
    "  \"time\", next_timestamp.cast(LongType) - timestamp.cast(LongType))\n",
    "clickStreamWithTimeDF.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculate aggregated page hits and time spent"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n",
    "\n",
    "clickStreamWithTimeDF.registerTempTable(\"tempData\")\n",
    "val clickStreamWithDateTimeDF = sqlContext.sql(\n",
    "  \"select eventId, eventType, cast(from_unixtime(timestamp) as date), \" +\n",
    "    \"ipaddress, sessionId, userId, pageUrl, browser, time \" +\n",
    "    \"from tempData\").withColumnRenamed(\n",
    "      \"CAST(from_unixtime(CAST(timestamp AS BIGINT), yyyy-MM-dd HH:mm:ss) AS DATE)\",\n",
    "      \"date\")\n",
    "// clickStreamWithDateTimeDF.show(5)\n",
    "\n",
    "clickStreamWithDateTimeDF.registerTempTable(\"ClickData\")\n",
    "val clicksDF = sqlContext.sql(\n",
    "  \"select pageURL, count(*) as page_hits, sum(time) as total_time \" +\n",
    "  \"from ClickData where eventType='pageView' group by pageURL\")\n",
    "clicksDF.show(5,false)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "clicksDF.registerTempTable(\"WebMetricsData\")\n",
    "val webMetricsDF = sqlContext.sql(\"select * from WebMetricsData\")\n",
    "webMetricsDF.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculate aggregated web metrics by product line, product, and feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "clicksDF.registerTempTable(\"WebMetricsDataTest\")\n",
    "val metricsQuery = \"\"\"\n",
    "  select\n",
    "    parse_URL(pageURL,'QUERY','product_line') as product_line, \n",
    "    Coalesce(parse_URL(pageURL,'QUERY','action'),'') as action,\n",
    "    Coalesce(parse_URL(pageURL,'QUERY','product'),'') as product, \n",
    "    Coalesce(parse_URL(pageURL,'QUERY','feature'),'') as feature, page_hits, total_time\n",
    "  from WebMetricsData\"\"\"\n",
    "val metricsQuery2 = \"\"\"\n",
    "  select\n",
    "    parse_URL(pageURL,'QUERY','product_line') as product_line, \n",
    "    parse_URL(pageURL,'QUERY','action') as action,\n",
    "    parse_URL(pageURL,'QUERY','product') as product, \n",
    "  from WebMetricsData\"\"\"\n",
    "val metricsQuery3 = \"\"\"\n",
    "  select parse_URL(pageURL,'QUERY','product_line') as product_line\n",
    "  from WebMetricsDataTest\"\"\"\n",
    "val webMetricsDF3 = sqlContext.sql(metricsQuery3).filter($\"product_line\".isNotNull).sort($\"product_line\".desc)\n",
    "webMetricsDF3.show(5)\n",
    "val webMetricsDF = sqlContext.sql(metricsQuery).filter($\"product_line\".isNotNull).sort($\"product_line\".desc)\n",
    "webMetricsDF.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregated web metrics for all product lines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val productlineMetrics = webMetricsDF.\n",
    "  select(\"product_line\",\"page_hits\",\"total_time\").\n",
    "  groupBy(\"product_line\").agg(sum(\"page_hits\"), sum(\"total_time\")).\n",
    "  withColumnRenamed(\"sum(page_hits)\",\"page_hits\").\n",
    "  withColumnRenamed(\"sum(total_time)\",\"total_time\").\n",
    "  sort($\"page_hits\".desc)\n",
    "   \n",
    "productlineMetrics.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "%%brunel data('productlineMetrics') \n",
    "bar at(0,0,50,50) title(\"Page Views by Product Line\")\n",
    "  x(product_line) y(page_hits)\n",
    "  tooltip(#all) color(product_line) legends(none)\n",
    "  axes(x:'product lines',y:'page views') sort(page_hits) interaction(select)|\n",
    "treemap at(60,5,100,45)\n",
    "  sort(page_hits) size(page_hits) color(product_line) label(product_line) legends(none)\n",
    "  tooltip(\"page views: \",page_hits) opacity(#selection) | \n",
    "bar at(0,50,50,100) title(\"Total Time by Product Line\")\n",
    "  x(product_line) y(total_time)\n",
    "  tooltip(#all) color(product_line) legends(none)\n",
    "  axes(x:'product lines',y:'total time') sort(page_hits) interaction(select)|\n",
    "treemap at(60,55,100,95)\n",
    "  sort(page_hits) size(total_time) color(product_line) label(product_line) legends(none)\n",
    "  tooltip(\"time on page (sec): \",total_time) opacity(#selection)\n",
    ":: width=1000, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregated web metrics for smart phones"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "val productMetrics = webMetricsDF.\n",
    "  select(\"product_line\",\"product\",\"page_hits\",\"total_time\").\n",
    "  filter($\"action\" === \"details\").filter($\"product_line\" === \"smartphones\").\n",
    "  groupBy(\"product_line\",\"product\").agg(sum(\"page_hits\"), sum(\"total_time\")).\n",
    "  withColumnRenamed(\"sum(page_hits)\",\"page_hits\").\n",
    "  withColumnRenamed(\"sum(total_time)\",\"total_time\")\n",
    "productMetrics.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel data('productMetrics') \n",
    "bar at(0,0,50,50) title(\"Page Views by Smart Phone\")\n",
    "  x(product) y(page_hits)\n",
    "  tooltip(page_hits,product) color(product) legends(none)\n",
    "  axes(x:'smart phones',y:'page views') sort(page_hits) interaction (select)|\n",
    "treemap at(60,5,100,45)\n",
    "  sort(page_hits) size(page_hits) color(product) label(product) legends(none)\n",
    "  tooltip(\"page views: \",page_hits) opacity(#selection) | \n",
    "bar at(0,50,50,100)\n",
    "  title(\"Total Time by Smart Phone\")\n",
    "  x(product) y(total_time)\n",
    "  color(product) label(product) tooltip(\"time on page (sec): \",total_time)\n",
    "  legends(none) sort(page_hits) interaction(select)|\n",
    "treemap at(60,55,100,95)\n",
    "  sort(page_hits) size(total_time) color(product) label(product) legends(none)\n",
    "  tooltip(\"time on page (sec): \",total_time) opacity(#selection)\n",
    " :: width=1000, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregated web metrics for smart phone features "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "val featureMetrics = webMetricsDF.\n",
    "  select(\"product\", \"feature\", \"page_hits\", \"total_time\").\n",
    "  filter($\"action\" === \"details\").\n",
    "  filter($\"product\" === \"A-phone\").\n",
    "  filter(\"feature != ''\").\n",
    "  groupBy(\"product\",\"feature\").agg(sum(\"page_hits\"), sum(\"total_time\")).\n",
    "  withColumnRenamed(\"sum(page_hits)\",\"page_hits\").\n",
    "  withColumnRenamed(\"sum(total_time)\",\"total_time\")\n",
    "\n",
    "featureMetrics.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%%brunel data('featureMetrics') \n",
    "bar title(\"Web Metrics by Feature\")\n",
    "  x(feature) y(page_hits)\n",
    "  tooltip(feature,page_hits) color(feature) legends(none)\n",
    "  axes(x:'A-phone features',y:'page views') sort(page_hits) interaction(select)|\n",
    "stack polar bar\n",
    "  y(total_time) color(feature) label(feature)\n",
    "  tooltip(\"time on page (sec): \",total_time)\n",
    "  legends(none) sort(page_hits) opacity(#selection)\n",
    ":: width=1000, height=300"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Detailed web metrics for user 'David'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val userClicksQuery =\"\"\"\n",
    "  select pageURL, year(date) as year, month(date) as month, weekofyear(date) as week,\n",
    "    day(date) as day, date_format(date, 'E') as dayofweek,\n",
    "    count(*) as page_hits, sum(time) as total_time\n",
    "  from ClickData\n",
    "  where eventType='pageView' and userId='datkins' group by pageURL, date\"\"\"\n",
    "val userClicksDF = sqlContext.sql(userClicksQuery)        \n",
    "userClicksDF.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// Build user web metrics by product_line, products and feature browses\n",
    "userClicksDF.registerTempTable(\"UserWebMetricsData\")\n",
    "val metricsQuery = \"\"\"\n",
    "  select month, week, day, dayofweek,\n",
    "    parse_URL(pageURL,'QUERY','product_line') as product_line, \n",
    "    Coalesce(parse_URL(pageURL,'QUERY','action'),'') as action,\n",
    "    Coalesce(parse_URL(pageURL,'QUERY','product'),'') as product, \n",
    "    Coalesce(parse_URL(pageURL,'QUERY','feature'),'') as feature,\n",
    "    page_hits, total_time\n",
    "  from UserWebMetricsData\n",
    "  where year = '2017'\"\"\"\n",
    "\n",
    "val userWebMetricsDF = sqlContext.sql(metricsQuery).filter($\"product_line\".isNotNull)\n",
    "userWebMetricsDF.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// Metrics for most recent week\n",
    "val weekMetricsDF = userWebMetricsDF.\n",
    "  groupBy(\"dayofweek\", \"day\", \"product_line\", \"action\", \"product\", \"feature\", \"page_hits\", \"total_time\").\n",
    "  max(\"week\")\n",
    "weekMetricsDF.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel data('weekMetricsDF') \n",
    "title(\"David's Browsing by Day\")\n",
    "x(day) y(page_hits)\n",
    "stack bar\n",
    "  sum(page_hits) color(product_line) tooltip(#all)\n",
    "  axes(x:7,y:'page views') legends(none) interaction(select)|\n",
    "stack polar bar\n",
    "  y(total_time) color(product_line) label(product)\n",
    "  tooltip(\"<b>day of week: \", dayofweek,\n",
    "          \"<p><u>day of month: \", day,\n",
    "          \"</u></b><p><i>product line: \", product_line,\n",
    "          \"</i><p>product: \", product)\n",
    "  opacity(#selection)\n",
    ":: width=1000, height=300"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## Insight summary from clickstream analysis\n",
    "\n",
    "1. Aggregated web metrics of recent months highlights significant interest in smart phones with A-phones leading the pack. \n",
    "2. User 'David' is a repeat visitor and has explored smart phones multiple times in recent days along with computers and headphones. \n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "file_extension": ".scala",
   "name": "scala",
   "version": "2.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
